package org.apache.flink.runtime.rest.handler.taskmanager;

import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;

import javax.annotation.Nonnull;

import org.apache.flink.api.common.time.Time;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
import org.apache.flink.runtime.resourcemanager.ResourceProviderInfoWithSlots;
import org.apache.flink.runtime.resourcemanager.exceptions.UnknownResourceProviderException;
import org.apache.flink.runtime.rest.handler.HandlerRequest;
import org.apache.flink.runtime.rest.handler.RestHandlerException;
import org.apache.flink.runtime.rest.handler.resourcemanager.AbstractResourceManagerHandler;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
import org.apache.flink.runtime.rest.messages.MessageHeaders;
import org.apache.flink.runtime.rest.messages.taskmanager.ResourceProviderIdPathParameter;
import org.apache.flink.runtime.rest.messages.taskmanager.ResourceProviderMessageParameters;
import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerDetailsInfo;
import org.apache.flink.runtime.rpc.retriever.GatewayRetriever;
import org.apache.flink.runtime.webmonitor.RestfulGateway;
import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
import org.apache.flink.util.ExceptionUtils;

/** Handler which serves detailed TaskManager information. */
public class ResourceProviderDetailsHandler
        extends AbstractResourceManagerHandler<
                RestfulGateway,
                EmptyRequestBody,
                TaskManagerDetailsInfo,
                ResourceProviderMessageParameters> {

    public ResourceProviderDetailsHandler(
            GatewayRetriever<? extends RestfulGateway> leaderRetriever,
            Time timeout,
            Map<String, String> responseHeaders,
            MessageHeaders<EmptyRequestBody, TaskManagerDetailsInfo, ResourceProviderMessageParameters>
                    messageHeaders,
            GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever) {
        super(
                leaderRetriever,
                timeout,
                responseHeaders,
                messageHeaders,
                resourceManagerGatewayRetriever);

    }

    @Override
    protected CompletableFuture<TaskManagerDetailsInfo> handleRequest(
            @Nonnull HandlerRequest<EmptyRequestBody> request,
            @Nonnull ResourceManagerGateway gateway)
            throws RestHandlerException {
        final ResourceID providerManagerResourceId =
                request.getPathParameter(ResourceProviderIdPathParameter.class);

        CompletableFuture<ResourceProviderInfoWithSlots> resourceProviderInfoWithSlotsFuture =
                gateway.requestTaskManagerDetailsInfo(providerManagerResourceId, timeout);

        return resourceProviderInfoWithSlotsFuture
                .thenApply(
                        (taskManagerInfoWithSlots) -> {

                            return new TaskManagerDetailsInfo(
                                    taskManagerInfoWithSlots);
                        })
                .exceptionally(
                        (Throwable throwable) -> {
                            final Throwable strippedThrowable =
                                    ExceptionUtils.stripExecutionException(throwable);

                            if (strippedThrowable instanceof UnknownResourceProviderException) {
                                throw new CompletionException(
                                        new RestHandlerException(
                                                "Could not find Resource Provider "
                                                        + providerManagerResourceId
                                                        + '.',
                                                HttpResponseStatus.NOT_FOUND,
                                                strippedThrowable));
                            } else {
                                throw new CompletionException(strippedThrowable);
                            }
                        });
    }

}
